package rdb;

import exception.CustomException;
import rdb.model.*;

import java.io.*;
import java.nio.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

/*
 rdb 文件格式：http://rdb.fnordig.de/file_format.html#crc64-checksum
 go rdb 文件实现：https://www.cnblogs.com/Finley/p/16251360.html
 */
public class RDBParser {
    private File rdbFile;
    private BufferedInputStream buffer;

    public RDBParser(String rdbFileName) {
        rdbFile = new File(rdbFileName);
        try {
            buffer = new BufferedInputStream(new FileInputStream(rdbFile));
        } catch (FileNotFoundException fe) {
            fe.printStackTrace();
        }
    }

    public boolean checkSum() {
        // 读取除 末尾 8个字节外的所有字节，并做校验
        long fileSize = rdbFile.length();
        try {
            BufferedInputStream buf = new BufferedInputStream(new FileInputStream(rdbFile));
            byte[] content = buf.readNBytes((int)fileSize-8);
            byte[] csBytes = buf.readNBytes(8);
            long realSum = CRC.calculateCRC(CRC.Parameters.CRC64ISO, content);
            long expectedSum = ByteBuffer.wrap(csBytes).order(ByteOrder.LITTLE_ENDIAN).getLong();
            System.out.println(realSum);

        } catch (IOException e) {

        }

        return false;
    }

    public boolean checkHeader() {
        byte[] redisByte = new byte[5];
        byte[] versionByte = new byte[4];
        try {
            int res = buffer.read(redisByte);
            if (res == -1) {
                return false;
            }
            if (!Arrays.equals(redisByte, RDBConstant.MAGIC_NUMBER)) {
                return false;
            }

            res = buffer.read(versionByte);
            if (res == -1) {
                return false;
            }
            int version = Integer.parseInt(new String(versionByte));
            return version >= RDBConstant.MIN_VERSION && version <= RDBConstant.MAX_VERSION;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    public List<byte[]> readList() {
        // 列表对象的编码可以是 ziplist 或 linkedlist
        // linkedlist 每个节点都是一个字符串对象，这里只解析 linkedlist 编码的的列表，ziplist 有其他方法解析
        Result lengthResult = readLength();
        int len = (int)lengthResult.dataLength;
        ArrayList<byte[]> values = new ArrayList<>();
        for (int i=0;i<len;i++) {
            values.add(readString());
        }
        return values;
    }

    public byte[][] readSet() {
        List<byte[]> temp = readList();
        byte[][] ret = new byte[temp.size()][];
        int i = 0;
        for(byte[] b : temp) {
            ret[i] = b;
            i++;
        }
        return ret;
    }

    public byte[][] readIntSet() throws IOException {
        byte[] buf = readString();
        byte [][] ret = null;
        int cur = 0;
        // 前 4 个字节表示每个元素在几个字节，2, 4, 8 个字节（小端序）
        int intSize = ((int) buf[cur++] & 0xff) |
                      ((int) buf[cur++] & 0xff) << 8 |
                      ((int) buf[cur++] & 0xff) << 16 |
                      ((int) buf[cur++] & 0xff) << 24;

        // 后 4 个字节表示元素个数（小端序）
        int setSize = ((int) buf[cur++] & 0xff) |
                      ((int) buf[cur++] & 0xff) << 8  |
                      ((int) buf[cur++] & 0xff) << 16 |
                      ((int) buf[cur++] & 0xff) << 24;
        ret = new byte[setSize][];
        ByteBuffer bb = ByteBuffer.wrap(buf, cur, buf.length-cur).order(ByteOrder.LITTLE_ENDIAN);
        if (intSize == 2) {
            ShortBuffer sb = bb.asShortBuffer();
            for (int i=0;i<sb.limit();i++) {
                ret[i] = Short.toString(sb.get()).getBytes(StandardCharsets.UTF_8);
            }
        } else if (intSize == 4) {
            IntBuffer ib = bb.asIntBuffer();
            for (int i=0;i<ib.limit();i++) {
                ret[i] = Integer.toString(ib.get()).getBytes(StandardCharsets.UTF_8);
            }
        } else if (intSize == 8) {
            LongBuffer lb = bb.asLongBuffer();
            for (int i=0;i<lb.limit();i++) {
                ret[i] = Long.toString(lb.get()).getBytes(StandardCharsets.UTF_8);
            }
        }

        return ret;
    }

    public HashMap<String, byte[]> readHashmap() {
        Result sizeResult = readLength();
        int size = (int) sizeResult.dataLength;
        HashMap<String, byte[]> hashMap = new HashMap<>();
        for (int i=0;i<size;i++) {
            byte[] field = readString();
            byte[] value = readString();
            hashMap.put(new String(field), value);
        }
        return hashMap;
    }

    public byte[][] readZipList(byte[] buf) throws IOException {
        // zlbytes 4B + zltail 4B = 8B, 直接跳过前面 8 个字节
        int cur = 8;

        // zllen 2B
        // 这里是小端序，内存低地址存的数字的低位，内存高地址存的是数字的高位
        int zlLen = (buf[cur++] & 0xff) | (buf[cur++] & 0xff) << 8;
        byte[][] ret = new byte[zlLen][];
        Statement statement = new Statement(buf, cur);
        for (int j=0;j<zlLen;j++) {
            ret[j] = readZipEntry(statement);
        }

        return ret;
    }

    public byte[] readZipEntry(Statement statement) throws IOException {
        // 先读一个字节
        int flag = statement.buf[statement.cur++] & 0xff;
        if (flag == 0xFE) {
            // 表示 prevLen 由 5 个字节表示，第一个字节是固定的 0xFE, 后四个字节表达 prevLen 的长度
            // 读 4 个字节, 相当于跳过 4 个字节
            statement.cur += 4;
        }
        // prevLen 由 一个字节表示, flag 就是 prevLen 的长度
        // 接着再读一个字节， encoding
        // encoding 的第一个字节的前两位为 11 时表示 entry-data 为整数，其它情况表示 entry-data 为字符串。
        // 先取前两位
        int code = statement.buf[statement.cur++] & 0xff;
        // 无符号右移
        int type = code >>> 6;
        if (type == 3) {
            byte[] contentBytes = null;
            long content = 0;
            // entry-data 为整数
            switch (code) {
                case RDBConstant.ZIP_INT_8B:
                    content = (long) statement.buf[statement.cur++] & 0xff;
                    return String.valueOf(content).getBytes(StandardCharsets.UTF_8);
                case RDBConstant.ZIP_INT_16B:
                    content = ((long) statement.buf[statement.cur++] & 0xff) |
                              ((long) statement.buf[statement.cur++] & 0xff) << 8;
                    return String.valueOf(content).getBytes(StandardCharsets.UTF_8);
                case RDBConstant.ZIP_INT_24B:
                    content = ((long) statement.buf[statement.cur++] & 0xff) |
                              ((long) statement.buf[statement.cur++] & 0xff) << 8 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 16;
                    return String.valueOf(content).getBytes(StandardCharsets.UTF_8);
                case RDBConstant.ZIP_INT_32B:
                    content = ((long) statement.buf[statement.cur++] & 0xff) |
                              ((long) statement.buf[statement.cur++] & 0xff) << 8 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 16 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 24;
                    return String.valueOf(content).getBytes(StandardCharsets.UTF_8);
                case RDBConstant.ZIP_INT_64B:
                    content = ((long) statement.buf[statement.cur++] & 0xff)       |
                              ((long) statement.buf[statement.cur++] & 0xff) << 8  |
                              ((long) statement.buf[statement.cur++] & 0xff) << 16 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 24 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 32 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 40 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 48 |
                              ((long) statement.buf[statement.cur++] & 0xff) << 56;

                    return String.valueOf(content).getBytes(StandardCharsets.UTF_8);

            }
            if (code >>> 4 == RDBConstant.ZIP_INT_4B) {
                //  使用这个编码的节点没有 content 属性，编码本身后四位保存了一个 0-12的值
                        /*
                        为什么要减一呢？
                        1111xxxx 用于表示范围在 [0, 12] 内的整数。
                        它的低 4 位取值 0000 或 1111 会与 int24 encoding 或者 zlend 发生冲突，
                        所以低 4 位只能在 [0001, 1101] 内取值。小整数会先加 1 然后再存入 encoding 的低 4 位中，
                        比如整数 0 实际上存储为 0001、1 实际存储为 0010、 12 存储为 1101。
                         */
                content = (code & 0x0F) - 1;
                return String.valueOf(content).getBytes(StandardCharsets.UTF_8);
            }
        } else {
            // entry-data 为字符串
            byte[] contentBytes = null;
            byte[] nextBytes = null;
            int bytesLen = 0;
            switch (type) {
                case RDBConstant.ZIP_STR_06B:
                    bytesLen = code & 0x3F;
                    contentBytes = new byte[bytesLen];
                    System.arraycopy(statement.buf, statement.cur, contentBytes, 0, bytesLen);
                    statement.cur += bytesLen;
                    break;
                case RDBConstant.ZIP_STR_14B:
                    // 这里是大端序
                    bytesLen = (code & 0x3F) << 8 | ((int)statement.buf[statement.cur++] & 0xff);
                    contentBytes = new byte[bytesLen];
                    System.arraycopy(statement.buf, statement.cur, contentBytes, 0, bytesLen);
                    statement.cur += bytesLen;
                    break;
                case RDBConstant.ZIP_STR_32B:
                    bytesLen = ((int) statement.buf[statement.cur++] & 0xff) << 24 |
                               ((int) statement.buf[statement.cur++] & 0xff) << 16 |
                               ((int) statement.buf[statement.cur++] & 0xff) << 8  |
                               ((int) statement.buf[statement.cur++] & 0xff);
                    contentBytes = new byte[bytesLen];
                    System.arraycopy(statement.buf, statement.cur, contentBytes, 0, bytesLen);
                    statement.cur += bytesLen;
                    break;
            }
            return contentBytes;
        }
        return null;
    }

    public List<byte[]> readQuickList() throws IOException {
        // 先获取 quicklist 中包含几个 zip list
        int size = (int) readLength().dataLength;
        ArrayList<byte[]> ret = new ArrayList<>();
        for (int i=0;i<size;i++) {
            // 每个 zip list 是转换成字符串后，再写入文件的
            byte[] ziplistStrBytes = readString();
            byte[][] page = readZipList(ziplistStrBytes);
            // 把每一个 zipEntry 都加入到 ret 中
            ret.addAll(Arrays.asList(page));
        }
        return ret;
    }

    public HashMap<String, byte[]> readHashZipList() throws IOException {
        // 对于 type 是 REDIS_HASH_ZIP_LIST, value 是一个 zip list，而且这个 zip list 是以字符串的形式保存的
        // 所以先读取字符串对象
        byte[][] entries = readZSetZipList();
        HashMap<String, byte[]> vals = new HashMap<>();
        for (int i=0;i<entries.length;i+=2) {
            vals.put(new String(entries[i]), entries[i+1]);
        }
        return vals;
    }

    public byte[][] readZSetZipList() throws IOException {
        // 对于 type 是 REDIS_ZSET_ZIP_LIST, value 是一个 zip list，而且这个zip list 是以字符串的形式保存的
        // 所以先读取字符串对象
        byte[] zipListStrBytes = readString();
        //  再来解析这个 zip list byte数组
        return readZipList(zipListStrBytes);
    }

    public byte[][] readListZipList() throws IOException {
        // 和 RDBWriter writeListZipList 对应
        // 对于 type 是 REDIS_LIST_ZIP_LIST, value 是一个 zip list，而且这个zip list 是以字符串的形式保存的
        // 所以先读取字符串对象
        return readZSetZipList();
    }

    public BaseObject readObject(int type, BaseObject baseObject) throws CustomException, IOException {
        try {
            switch (type) {
                case RDBConstant.TYPE_STRING:
                    byte[] val = readString();
                    baseObject.setType(ValueType.STRING);
                    return new StringObject(baseObject, val);
                case RDBConstant.TYPE_LIST:
                    List<byte[]> list = readList();
                    baseObject.setType(ValueType.LIST);
                    return new ListObject(baseObject, list);
                case RDBConstant.TYPE_SET:
                    byte[][] set = readSet();
                    baseObject.setType(ValueType.SET);
                    return new SetObject(baseObject, set);
                case RDBConstant.TYPE_SET_INT_SET:
                    byte[][] intSet = readIntSet();
                    baseObject.setType(ValueType.SET_INTSET);
                    return new SetObject(baseObject, intSet);
                case RDBConstant.TYPE_HASH:
                    HashMap<String, byte[]> hashMap = readHashmap();
                    baseObject.setType(ValueType.HASH);
                    return new HashObject(baseObject, hashMap);
                case RDBConstant.TYPE_LIST_ZIP_LIST:
                    byte[][] listZipListVals = readListZipList();
                    baseObject.setType(ValueType.LIST_ZIPLIST);
                    return new ListObject(baseObject, Arrays.asList(listZipListVals));
                case RDBConstant.TYPE_LIST_QUICK_LIST:
                    List<byte[]> vals = readQuickList();
                    baseObject.setType(ValueType.LIST_QUICK_LIST);
                    return new ListObject(baseObject, vals);
                case RDBConstant.TYPE_HASH_ZIP_LIST:
                    HashMap<String, byte[]> hashVals = readHashZipList();
                    baseObject.setType(ValueType.HASH_ZIPLIST);
                    return new HashObject(baseObject, hashVals);
                case RDBConstant.TYPE_ZSET_ZIP_LIST:
                    byte[][] zipListVals = readZSetZipList();
                    baseObject.setType(ValueType.ZSET_ZIPLIST);
                    return new ZSetObject(baseObject, zipListVals);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    public List<BaseObject> parse() {
        if (!checkHeader()) {
            return null;
        }
        int dbIndex = 0;
        long expireMs = 0;
        ArrayList<BaseObject> ret = new ArrayList<>();
        try {
            while (true) {
                // 先读一个字节, 这个可能是 op code，也可能是数据区，代表某种类型的数据结构
                byte[] b = buffer.readNBytes(1);
                int code = b[0] & 0xff;
                if (code == RDBConstant.OP_CODE_EOF) {
                    break;
                } else if (code == RDBConstant.OP_CODE_SELECT_DB) {
                    Result result = readLength();
                    dbIndex = (int) result.dataLength;
                    continue;
                } else if (code == RDBConstant.OP_CODE_EXPIRE_TIME) {
                    b = buffer.readNBytes(4);
                    expireMs = convert(b, ByteOrder.LITTLE_ENDIAN, b.length);
                    // 统一转换成毫秒
                    expireMs *= 1000;
                    continue;
                } else if (code == RDBConstant.OP_CODE_EXPIRE_TIME_MS) {
                    b = buffer.readNBytes(8);
                    expireMs = convert(b, ByteOrder.LITTLE_ENDIAN, b.length);
                    continue;
                } else if (code == RDBConstant.OP_CODE_RESIZE_DB) {
                    // 获取数据库中 key 的数量，以及具有过期时间的 key 的数量
                    long keyCount, ttlCount;
                    Result result = readLength();
                    keyCount = result.dataLength;
                    result = readLength();
                    ttlCount = result.dataLength;
                    continue;
                } else if (code == RDBConstant.OP_CODE_AUX) {
                    // The op code is followed by two Redis Strings, representing the key and value of a setting
                    // 每个 redis string 是按照 length-encoding 编码的
                    readString();
                    readString();
                    continue;
                } else if (code == RDBConstant.OP_CODE_FREQ) {
                    buffer.readNBytes(1);
                    continue;
                } else if (code == RDBConstant.OP_CODE_IDLE) {
                    readLength();
                    continue;
                }
                // 如果不是上述特殊 code，那就是读到数据了，数据开始时，是 key，而 key 永远是 string
                byte[] key = readString();
                BaseObject baseObject = new BaseObject();
                BaseObject obj = readObject(code, baseObject);
                obj.setKey(new String(key));
                obj.setDbIndex(dbIndex);
                obj.setExpiration(expireMs);
                ret.add(obj);
            }
            return ret;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                buffer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /*
    Length Encoding 是一种可变长度的无符号整型编码，因为通常被用来存储字符串长度、列表长度等长度数据所以被称为 Length Encoding.
    如果前两位是 00 那么下面剩下的 6 位就表示具体长度
    如果前两位是 01 那么会再读取一个字节的数据，加上前面剩下的6位，共14位用于表示具体长度
    如果前两位是 10 如果剩下的 6 位都是 0 那么后面 32 个位表示具体长度。如果剩下的 6 位为 000001, 那么后面的 64 位表示具体长度。（注意有些较老的文章没有提及 64 位的 Length Encoding）
    如果前两位是 11 表示为使用字符串存储整数的特殊编码，
     */
    public Result readLength() {
        Result result = new Result();
        long ret = 0;
        boolean isSpecial = false;
        try {
            // 先读一个字节
            byte[] bs = buffer.readNBytes(1);
            // 取前 2 位
            int firstTwo = (bs[0] & 0xc0) >> 6;
            switch (firstTwo) {
                case RDBConstant.LEN_6_BIT:
                    ret = (bs[0] & 0x3f);
                    break;
                case RDBConstant.LEN_14_BIT:
                    byte[] nextOne = buffer.readNBytes(1);
                    ret = ((bs[0] & 0x3f) << 8) | nextOne[0];
                case RDBConstant.LEN_32_OR_64_BIT:
                    // 后 6 位是 0，那个整个字节就是 1000 0000 = 0x80
                    // 因为 java byte 类型取值范围是 -128 - 127，但是 0x80 = 128, 超了
                    // java byte 取值范围讲解：https://www.how234.com/eamyrzccc.html
                    long code = convert(bs, ByteOrder.BIG_ENDIAN, 1);
                    if (code == RDBConstant.LEN_32_BIT) {
                        byte[] next4Bytes = buffer.readNBytes(4);
                        ret = convert(next4Bytes, ByteOrder.BIG_ENDIAN, next4Bytes.length);
                    } else if (code == RDBConstant.LEN_64_BIT) {
                        // 后 6 位是 1，那么整个字节就是 1000 0001 = 0x81
                        byte[] next8Bytes = buffer.readNBytes(8);
                        ret = convert(next8Bytes, ByteOrder.BIG_ENDIAN, next8Bytes.length);
                    }
                case RDBConstant.LEN_SPECIAL:
                    isSpecial = true;
                    // 直接返回后 6 位，如果是特殊编码，则第一个字节剩下的 6 位会表示具体编码方式
                    ret = bs[0] & 0x3f;
            }
            result.isSpecial = isSpecial;
            result.dataLength = ret;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    public byte[] readString() {
        Result result = readLength();
        long ret = 0;
        byte[] b = null;
        try {
            if (result.isSpecial) {
                if (result.dataLength == RDBConstant.STRING_ENCODE_INT_8) {
                    b = buffer.readNBytes(1);
                } else if (result.dataLength == RDBConstant.STRING_ENCODE_INT_16) {
                    b = buffer.readNBytes(2);
                } else if (result.dataLength == RDBConstant.STRING_ENCODE_INT_32) {
                    b = buffer.readNBytes(4);
                } else if (result.dataLength == RDBConstant.STRING_ENCODE_LZF) {
                    // 如果是 zip list，那么它一定是以一个字符串的形式存入，要么是特殊编码的 lzf，要么是普通编码的字符串
                    System.out.println("read lzf");
                    return readLZF();
                }
                if (b != null) {
                    ret = convert(b, ByteOrder.LITTLE_ENDIAN, b.length);
                }
                return String.valueOf(ret).getBytes(StandardCharsets.UTF_8);
            }

            //  如果不是特殊编码，就是普通编码，直接读取 result.dataLength 个字节
            byte[] res = buffer.readNBytes((int) result.dataLength);
            return res;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private byte[] readLZF() throws IOException {
        Result inResult = readLength();
        Result outResult = readLength();

        byte[] src = buffer.readNBytes((int)inResult.dataLength);
        byte[] dest = new byte[(int) outResult.dataLength];
        expand(src, dest);
        return dest;
    }

    public static void expand(byte[] src, byte[] dest) {
        int srcPos = 0;
        int destPos = 0;
        do {
            int ctrl = src[srcPos++] & 0xff;
            if (ctrl < 32) {
                // literal run of length = ctrl + 1,
                ctrl++;
                // copy to output and move forward this many bytes
                System.arraycopy(src, srcPos, dest, destPos, ctrl);
                destPos += ctrl;
                srcPos += ctrl;
            } else {
        /* back reference
           the highest 3 bits are the match length */
                int len = ctrl >> 5;
                // if the length is maxed, add the next byte to the length
                if (len == 7) {
                    len += src[srcPos++] & 0xff;
                }
        /* minimum back-reference is 3 bytes,
           so 2 was subtracted before storing size */
                len += 2;

        /* ctrl is now the offset for a back-reference...
           the logical AND operation removes the length bits */
                ctrl = -((ctrl & 0x1f) << 8) - 1;

                // the next byte augments/increases the offset
                ctrl -= src[srcPos++] & 0xff;

        /* copy the back-reference bytes from the given
           location in output to current position */
                ctrl += destPos;
                for (int i = 0; i < len; i++) {
                    dest[destPos++] = dest[ctrl++];
                }
            }
        } while (destPos < dest.length);
    }

    private long convert(byte[] bs, ByteOrder byteOrder, int bytes) {
        ByteBuffer byteBuffer = ByteBuffer.wrap(bs).order(byteOrder);
        if (bytes == 1) {
            int ret = byteBuffer.get();
            return ret + 256;
        } else if (bytes == 2) {
            short res = byteBuffer.getShort();
            return (long) res;
        } else if (bytes == 4) {
            int res = byteBuffer.getInt();
            return res;
        } else if (bytes == 8) {
            long res = byteBuffer.getLong();
            return res;
        }
        return -1;
    }

    /*
    https://blog.csdn.net/qq_25405729/article/details/123947708
    Java长整数long、整数int与字节集合byte[]互转
     */
    public static byte[] longToBytes(long data) {
        byte[] res = new byte[8];
        for(int i = 0; i < res.length; i++) {
            res[i] = (byte)(data >> 8*i);
        }
        return res;
    }

    public static long bytesToLong(byte[] args) {
        if(args == null  || args.length != 8)
            throw(new NullPointerException("非法集合或集合为空！"));

        long res = 0;
        for(int i = args.length - 1; i >= 0 ; i--) {
            res = (res << 8) | (args[i] & 0xff);
        }
        return res;
    }

    public static byte[] integerToBytes(int data) {
        byte[] res = new byte[4];
        for(int i = 0; i < res.length; i++) {
            res[i] = (byte)(data >> 8*i);
        }
        return res;
    }

    public static int bytesToInteger(byte[] args) {
        if(args == null || args.length != 4)
            throw(new NullPointerException("非法集合或集合为空！"));

        int res = 0;
        for(int i = args.length - 1; i >= 0 ; i--) {
            res = (res << 8) | (args[i] & 0xff);
        }
        return res;
    }

    public static void main(String[] args) {
        RDBParser rdbParser = new RDBParser("dump.rdb");
        List<BaseObject> ret = rdbParser.parse();
        for (BaseObject obj : ret) {
            System.out.printf("key: %s, values: %s", obj.getKey(), obj);
        }
    }


}

